package com.gome.ss.flowcrawl.extractor;

import java.util.List;
import java.util.Set;

import javax.jms.JMSException;

import org.apache.log4j.Logger;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.search.SearchType;
import org.elasticsearch.client.Client;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.index.query.QueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.facet.FacetBuilders;
import org.elasticsearch.search.facet.terms.TermsFacet;
import org.elasticsearch.search.facet.terms.TermsFacetBuilder;

import com.gome.ss.common.AppConfiguer;
import com.gome.ss.common.Utils;
import com.gome.ss.core.schedule.ThreadTaskBase;
import com.mongodb.BasicDBObject;
import com.mongodb.DBCollection;
import com.mongodb.DBCursor;
import com.mongodb.DBObject;

public class FlowCrawlExtractorDoIndexTask extends FlowCrawlExtractorTask implements ThreadTaskBase {
	private static Logger logger = Logger.getLogger(FlowCrawlExtractorDoIndexTask.class);
	
	public FlowCrawlExtractorDoIndexTask(FlowCrawlExtractor extractor) {
		super(extractor, true);
	}
	
	private void megerParent(DBCollection collection, DBObject childObj, DBObject tempObj) {
		if (tempObj.get("parentPage") == null)
			return ;
		
		String pid = (String) ((DBObject) tempObj.get("parentPage")).get("id");
		BasicDBObject query = new BasicDBObject();
		query.append("id", pid);
		DBCursor cursor = collection.find(query);
		if (cursor.size() <= 0)
			return ;
		
		DBObject pobj = cursor.next();
		
		Set<String> keys = pobj.keySet();
		for (String key : keys) {
			if (key.equals(FlowCrawlExtractorManager.extractDataFacetKey)) {
				List<String> cfacets = (List<String>) childObj.get(key);
				List<String> facets = (List<String>) pobj.get(key);
				for (String facet : facets) {
					if (!cfacets.contains(facet))
						cfacets.add(facet);
				}
				childObj.put(key, cfacets);
				continue;
			}
			
			if (childObj.containsField(key))
				continue;
			childObj.put(key, pobj.get(key));
		}
		
		megerParent(collection, childObj, pobj);
	}
	
	@Override
	public void execute() {
		FlowCrawlExtractorConfig config = extractor.getConfig();
		
		DBCollection collection = AppConfiguer.flowCrawlConfigMongo.getDB(AppConfiguer.FLOWCRAWL_DB_NAME)
				 .getCollection(extractor.getConfig().getName()+FlowCrawlExtractorManager.collectionNameSuffix);
		
		BasicDBObject query = new BasicDBObject();
		query.append("stepID", config.getStepID());
		DBCursor cursor = collection.find(query);
		
		int curpos = 0;
		
		logger.info(config.getName() + " prepare index data count " + cursor.count());
		while(cursor.hasNext() && !needStop) {
			DBObject obj = cursor.next();
			if (obj != null) {
				try {
					megerParent(collection, obj, obj);
					putQueue(obj);
					session.commit();
				} catch (JMSException e) {
					logger.error(Utils.getTrace(e));
				}
			}
			curpos++;
			if (curpos%100 == 0)
				logger.info(config.getName() + " prepare index " + curpos + " has end.");
		}
		
		logger.info(config.getName() + " prepare index " + curpos + " end.");
		extractor.getStatus().setProgress(100);
		
		if (config.getEndCallbackLink() != null && !config.getEndCallbackLink().equals("")) {
			Utils.sendGet(config.getEndCallbackLink());
		}
		
		needStop = false;
		try {
			producer.close();
			session.close();
			connection.close();
		} catch (JMSException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		extractor.getStatus().stop();
	}
	
	@Override
	public void stop() {
		logger.info(extractor.getConfig().getName() + " extract data stop.");
		needStop = true;
	}
	
//	public static void main(String[] args) {
//		
//		String indexName = "flowcrawl_extractor_index";
//		String newIndexName = "flowcrawl_extractor_index_v1";
//		String indexType = "flowcrawl_extractor_data";
//		Client client = ElasticsearchClientFactory.getClient();
//		
//		ElasticsearchClientFactory.createMapping(client, newIndexName, indexType);
//		
//		QueryBuilder query = QueryBuilders.termQuery("keyWord", "京东");
//		
//		query = QueryBuilders.matchAllQuery();
//		
//		System.out.println(query.toString());
//		
//		TermsFacetBuilder termsFacet1 = FacetBuilders.termsFacet("termF1").field("keyWord");
//		TermsFacetBuilder termsFacet2 = FacetBuilders.termsFacet("termF2").field("crawlName");
//		
//		SearchResponse response = client.prepareSearch(indexName)
//		  .setTypes(indexType)
//		  .setSearchType(SearchType.SCAN)
//		  .setSize(5)
//		  .setScroll(TimeValue.timeValueMinutes(5))
//		  .addFacet(termsFacet1)
//		  .addFacet(termsFacet2)
//		  .execute().actionGet(); //100 hits per shard will be returned for each scroll
//		//Scroll until no hits are returned
//		
//		System.out.println(response.getHits().totalHits());
//		
//		TermsFacet f1 = (TermsFacet) response.getFacets().facetsAsMap().get("termF1");
//		TermsFacet f2 = (TermsFacet) response.getFacets().facetsAsMap().get("termF2");
//		
//        System.out.println(f1.getTotalCount() + " " + f1.getOtherCount() + " " + f1.getMissingCount());
//        System.out.println(f2.getTotalCount() + " " + f2.getOtherCount() + " " + f2.getMissingCount());
//        
//        for (TermsFacet.Entry entry : f1) {
//            System.out.println("t1:" + entry.getTerm()); // Term
//            System.out.println("c1:" + entry.getCount()); // Doc count
//            System.out.println("----");
//        }
//        
//        for (TermsFacet.Entry entry : f2) {
//            System.out.println("t2:" + entry.getTerm()); // Term
//            System.out.println("c2:" + entry.getCount()); // Doc count
//            System.out.println("----");
//        }
//        
//	    for (SearchHit hit : response.getHits()) {
//	    	
//	    	String id = hit.getId();
//	    	client.prepareDelete(newIndexName, indexType, id).execute().actionGet();
//			
//			IndexResponse iresponse = client.prepareIndex(newIndexName, indexType)
//						.setSource(hit.getSource()).setId(id).execute().actionGet();
//			
//			logger.info("Do index " + iresponse.getId());
//	    }
//	    
//        while(true) {
//        	response = client.prepareSearchScroll(response.getScrollId())
//        	        .setScroll(TimeValue.timeValueMinutes(5))
//        	        .execute().actionGet();
//        	
//        	System.out.println(response.getHits().hits().length);
//        	
//        	if (response.getHits().hits().length <= 0)
//        		break;
//        	
//		    for (SearchHit hit : response.getHits()) {
//		    	
//		    	String id = hit.getId();
//		    	client.prepareDelete(newIndexName, indexType, id).execute().actionGet();
//				
//				IndexResponse iresponse = client.prepareIndex(newIndexName, indexType)
//							.setSource(hit.getSource()).setId(id).execute().actionGet();
//				
//				//logger.info("Do index " + iresponse.getId());
//		    }
//        }
//	}
}